package cn.xshi.gateway.route.client;

import cn.hutool.core.collection.CollectionUtil;
import cn.xshi.gateway.route.client.vo.ChannelEntity;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import cn.xshi.gateway.util.RouteInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.CommandLineRunner;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
 * @Desc 网关路由客户端连接
 * @Author 邓纯杰
 * @CreateTime 2012-12-12 12:12:12
 */
@Component
@Slf4j
public class GatewayNettyClient implements CommandLineRunner, ApplicationEventPublisherAware {

    private static final String JEHC_CLOUD_OMP = "jehc-cloud-omp";//网关动态路由服务Server

    public Map<String, ChannelEntity> channelFutureMap = new ConcurrentHashMap<>();//全局Channel维护

    @Resource
    RouteInfo routeInfo;

    @Resource
    private GatewayNettyClient nettyClient;

    private ApplicationEventPublisher applicationEventPublisher;

    @Override
    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    @Resource
    DiscoveryClient discoveryClient;

    Bootstrap bootstrap = null;

    @Override
    public void run(String... args) throws Exception {
        new Thread(new GatewayNettyClientWorker()).start();
    }

    /**
     *
     */
    class GatewayNettyClientWorker implements Runnable{
        public void run(){
            initChannel();
        }
    }

    /**
     * 初始化Bootstrap
     */
    public Bootstrap initBootstrap(EventLoopGroup group) {
        if (null == group) {
            group = new NioEventLoopGroup();
        }
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new GatewayNettyClientInitializer(nettyClient,routeInfo.getClientGroupId(),routeInfo.getClientId(),applicationEventPublisher));
        return bootstrap;
    }

    /**
     * 初始化服务端Channel
     * @return
     */
    public void initChannel() {

        setGateWayAddressList();//初始化连接服务端Server路由服务中心ip地址

        bootstrap = initBootstrap(null);

        if (!StringUtils.isEmpty(routeInfo.getHost())) {
            String[] hpsArray = routeInfo.getHost().split(",");
            if (null != hpsArray) {
                for (String hostTemp : hpsArray) {
                    if (!StringUtils.isEmpty(hostTemp)) {
                        bootstrap.remoteAddress(hostTemp, new Integer(routeInfo.getPort()));
                        //异步连接tcp服务端
                        ChannelFuture channelFuture = bootstrap.connect().addListener((ChannelFuture futureListener) -> {
                            final EventLoop eventLoop = futureListener.channel().eventLoop();
                            if (!futureListener.isSuccess()) {
                                //服务器未启动 连接tcp服务器不成功
                                log.info(hostTemp + "初次连接与路由服务端断开连接!在"+routeInfo.getRetryTime()+"s之后,尝试重连!");
                                //10秒后重连
                                eventLoop.schedule(() -> doConnect(bootstrap, hostTemp, routeInfo.getPort()), routeInfo.getRetryTime(), TimeUnit.SECONDS);
                            }else{
                                log.info(hostTemp + "初次连接与路由服务端成功!");
                            }
                        });
                        put(hostTemp,routeInfo.getPort(),channelFuture);
//                        channelFutureMap.put(hostTemp, new ChannelEntity(channelFuture,hostTemp,routeInfo.getPort(),channelFuture.isSuccess()));//存放
                    }
                }
            }
        }
    }

    /**
     * 重连服务端
     * @param bootstrap
     * @param host
     * @param port
     */
    public void doConnect(Bootstrap bootstrap,String host,Integer port) {
        try {
            if (bootstrap != null) {
                bootstrap.remoteAddress(host, port);
                ChannelFuture channelFuture = bootstrap.connect().addListener((ChannelFuture futureListener) -> {
                    final EventLoop eventLoop = futureListener.channel().eventLoop();
                    if (!futureListener.isSuccess()) {
                        log.info("服务端："+host+"，端口："+port + "服务器断线-----与服务端断开连接!在"+routeInfo.getRetryTime()+"s之后，尝试重连!");
                        eventLoop.schedule(() -> doConnect(bootstrap, host, port), routeInfo.getRetryTime(), TimeUnit.SECONDS);
                    }
                });
                put(host,port,channelFuture);
//                channelFutureMap.put(host+":"+port, new ChannelEntity(channelFuture,host,port,channelFuture.isSuccess()));//覆盖
            }
        } catch (Exception e) {
            log.error("客户端连接失败,{}" , e.getMessage());
        }
    }

    /**
     * 查询所有网关动态路由OMP模块中心服务所在物理机ip
     * @return
     */
    private void setGateWayAddressList(){
        try {
            List<ServiceInstance> serverAddressList = discoveryClient.getInstances(JEHC_CLOUD_OMP);
            if(CollectionUtil.isNotEmpty(serverAddressList)){//查询所有的客户端服务注册到nacos中心的时候 WEB服务自身所在的客户端IP
                List<String> list = new ArrayList<>();
                for(ServiceInstance serviceInstance : serverAddressList){
                    list.add(serviceInstance.getHost());
                }
                String result = list.stream().distinct().collect( Collectors.joining( "," ) );
                routeInfo.setHost(result);
            }
        }catch (Exception e){
            log.info("查询动态路由OMP服务所在ip异常：{}",e);
        }
    }

    /**
     * 动态加载
     * @param newAddress
     */
    public void dynamicsLoad(String newAddress){
        if (!StringUtils.isEmpty(newAddress)) {
            String[] newAddressArray = newAddress.split(",");
            if (null != newAddressArray) {
                for (String addressTemp : newAddressArray) {
                    if (!StringUtils.isEmpty(addressTemp)) {
                        if(null == bootstrap){
                            bootstrap = initBootstrap(null);
                        }
                        bootstrap.remoteAddress(addressTemp, routeInfo.getPort());
                        //异步连接tcp服务端
                        ChannelFuture channelFuture = bootstrap.connect().addListener((ChannelFuture futureListener) -> {
                            final EventLoop eventLoop = futureListener.channel().eventLoop();
                            if (!futureListener.isSuccess()) {
                                //服务器未启动 连接tcp服务器不成功
                                log.info(addressTemp + "动态增加节点，初次连接与动态路由中心服务断开连接!在"+routeInfo.getRetryTime()+"s之后,尝试重连!");
                                //10秒后重连
                                eventLoop.schedule(() -> doConnect(bootstrap, addressTemp, routeInfo.getPort()), routeInfo.getRetryTime(), TimeUnit.SECONDS);
                            }else{
                                log.info(addressTemp + "动态增加节点，初次连接与动态路由中心服务端成功!");
                            }
                        });
                        put(addressTemp,routeInfo.getPort(),channelFuture);
//                        channelFutureMap.put(addressTemp, new ChannelEntity(channelFuture,addressTemp,oauthClientConfig.getPort(),channelFuture.isSuccess()));//存放
                    }
                }
            }
        }
    }

    /**
     * 异常channel通道缓存
     * @param host
     */
    public void removeChannelFuture(String host){
        if(StringUtils.isEmpty(host)){
            log.warn("移除Channel缓存时，host不存在！");
            return;
        }
        if(CollectionUtil.isEmpty(channelFutureMap)){
            log.warn("移除Channel缓存时，channelFutureMap为空");
            return;
        }
        channelFutureMap.remove(host+":"+routeInfo.getPort());
    }

    /**
     * 存放
     * @param host
     * @param port
     * @param channelFuture
     */
    public void put(String host,Integer port,ChannelFuture channelFuture){
        if(CollectionUtil.isEmpty(channelFutureMap)){
            channelFutureMap = new ConcurrentHashMap<>();
        }
        channelFutureMap.put(host+":"+port, new ChannelEntity(channelFuture,host,port,channelFuture.isSuccess()));//覆盖
    }

    /**
     *
     * @param key
     */
    public ChannelEntity getChannelEntity(String key){
        if(StringUtils.isEmpty(key)){
            log.warn("查询Channel缓存时，Key不存在！");
            return null;
        }
        if(CollectionUtil.isEmpty(channelFutureMap)){
            log.warn("查询Channel缓存时，channelFutureMap为空");
            return null;
        }
        return channelFutureMap.get(key);
    }

    /**
     * 查询所有缓存通道
     * @return
     */
    public Map<String, ChannelEntity> getChannelEntityList(){
        return channelFutureMap;
    }
}
